package cn.jly.flink.custom;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * write data to kafka
 *
 * @author lanyangji
 */
public class KafkaUtils2 {
    private static final String BROKER_LIST = "hadoop101:9092,hadoop102:9092,hadoop103:9092";
    private static final String TOPIC = "student";

    private static void writeToKafka() throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", BROKER_LIST);
        props.put("zookeeper.connect", "hadoop101:2181");
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        props.put("auto.create.topics.enable", true);
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 100; i++) {
            Student student = new Student();
            student.setId(i);
            student.setName("tom" + i);
            student.setPassword("password" + i);
            student.setAge(18 + i);

            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC,JSON.toJSONString(student));
            producer.send(record);
            System.out.println("发送数据：" + JSON.toJSONString(student));
        }

        producer.flush();
    }

    public static void main(String[] args) throws Exception {
        writeToKafka();
    }
}
